package cn.xinfei.xdecision.pulsar.consumer;

import cn.xinfei.xdecision.pulsar.XPulsarClientFactory;
import cn.xinfei.xdecision.pulsar.model.XPulsarConsumerModel;
import cn.xinfei.xdecision.pulsar.util.PulsarUtils;
import com.dianping.cat.Cat;
import io.micrometer.core.instrument.Metrics;
import io.micrometer.core.instrument.Tags;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;

import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;

@Data
@Slf4j
@EqualsAndHashCode(callSuper = false)
public class XPulsarConsumerThread implements Runnable {
    private Method method;
    private Object obj;
    private volatile String topic;
    private int concurrentNum;
    private String subscriptionName;
    private volatile XPulsarConsumer<String> consumer;
    private volatile String consumerName;

    private volatile boolean stopConsumer = false;
    private volatile boolean batch = false;

    public XPulsarConsumerThread(Method method, Object obj, int concurrentNum) {
        this.method = method;
        this.obj = obj;
        this.concurrentNum = concurrentNum;
        XPulsarConsumerAnnotate annotation = method.getAnnotation(XPulsarConsumerAnnotate.class);
        this.batch = annotation.batch();

        String subscriptionName = annotation.subscriptionName();
        this.subscriptionName = PulsarUtils.getValueFromEnv(subscriptionName);
        String consumerName = annotation.consumerName();
        this.consumerName = PulsarUtils.getValueFromEnv(consumerName);

        String topic = annotation.topic();
        this.topic = PulsarUtils.getValueFromEnv(topic);
    }

    @Override
    public void run() {
        try {
            Thread.currentThread().setName(topic + "_" + concurrentNum);
            XPulsarConsumerModel xPulsarConsumerModel = new XPulsarConsumerModel(topic, subscriptionName, consumerName, concurrentNum);

            Message<String> message = null;
            Messages<String> messages = null;
            while (!stopConsumer) {
                try {
                    if (!getConsumer(xPulsarConsumerModel)) {
                        continue;
                    }

                    if (!batch) {
                        message = consumer.getConsumer().receive(200, TimeUnit.MILLISECONDS);
                    } else {
                        messages = consumer.getConsumer().batchReceive();
                    }
                    if (message == null && messages == null) {
                        Thread.sleep(200);
                        continue;
                    }else{
                        Thread.sleep(5);
                    }
                    if (message != null && message.getData() != null) {
                        Object[] args = buildArgs(message, messages, consumer);
                        Object invoke = method.invoke(obj, args);
                    } else if (messages != null && messages.size() > 0) {
                        Object[] args = buildArgs(message, messages, consumer);
                        Object invoke = method.invoke(obj, args);
                    } else {
                        Thread.sleep(200);
                    }

                } catch (Exception e) {
                    log.error("xpulsar consumer fail.msg={}", e.getMessage(), e);
                    Cat.logError(e);
                    Metrics.counter("xengine_pulsar_concumer_exception_total").increment();
                    if (message != null) {
                        consumer.getConsumer().negativeAcknowledge(message);
                    }
                    if (messages != null) {
                        consumer.getConsumer().negativeAcknowledge(messages);
                    }
                }
            }
            log.warn("xpulsar-consumer-thread is stopped.topic={}", topic);

        } catch (Exception e) {
            log.error("xpulsar-consumer-thread create fail.msg={}", e.getMessage(), e);
        }
    }


    private boolean getConsumer(XPulsarConsumerModel xPulsarConsumerModel) throws InterruptedException {
        if (consumer == null || consumer.isConnected() == false) {
            log.info("消费者实例启动重试,topic={},consumerName={}", topic, consumerName);
            consumer = XPulsarClientFactory.newConsumer(xPulsarConsumerModel);
            if (consumer != null && consumer.isConnected()) {
                this.consumerName = consumer.getConsumer().getConsumerName();
                String subscription = consumer.getConsumer().getSubscription();
                monitor(xPulsarConsumerModel);
                log.info("消费者实例启动成功,topic={},consumerName={},subscription={}", topic, consumerName, subscription);
            }
        }
        if (consumer != null && consumer.isConnected()) {
//            String consumerName = consumer.getConsumer().getConsumerName();
//            String subscription = consumer.getConsumer().getSubscription();
//            log.info("消费者实例启动成功,topic={},consumerName={},subscription={}", topic, consumerName, subscription);
//            Thread.currentThread().setName(topic + "_" + concurrentNum + "_" + consumerName);

            return true;
        }
        Thread.sleep(1000);
        return false;
    }

    private void monitor(XPulsarConsumerModel xPulsarConsumerModel) {
        try {
            ConsumerStats stats = consumer.getStats();
            Integer msgNumInReceiverQueue = stats.getMsgNumInReceiverQueue();
            long totalMsgsReceived = stats.getTotalMsgsReceived();
            long totaBatchReceivedFailed = stats.getTotaBatchReceivedFailed();
            long totalAcksSent = stats.getTotalAcksSent();
            long totalAcksFailed = stats.getTotalAcksFailed();
            double rateMsgsReceived = stats.getRateMsgsReceived();
            Tags tags = Tags.of("topic", xPulsarConsumerModel.getTopic());
            Metrics.gauge("pulsar_queue_total", tags, stats, ConsumerStats::getMsgNumInReceiverQueue);
            Metrics.gauge("pulsar_receive_total", tags, stats, ConsumerStats::getTotalMsgsReceived);
            Metrics.gauge("pulsar_receive_fail", tags, stats, ConsumerStats::getTotaBatchReceivedFailed);
            Metrics.gauge("pulsar_ack_send", tags, stats, ConsumerStats::getTotalAcksSent);
            Metrics.gauge("pulsar_ack_fail", tags, stats, ConsumerStats::getTotalAcksFailed);
            Metrics.gauge("pulsar_receive_rate", tags, stats, ConsumerStats::getRateMsgsReceived);
//                    log.info("topic={},stats={}",topic, JsonUtils.toJsonString(stats));
            log.info("topic={},msgNumInReceiverQueue={},totalMsgsReceived={},totaBatchReceivedFailed={},totalAcksSent={},totalAcksFailed={},rateMsgsReceived={}",
                    xPulsarConsumerModel.getTopic(), msgNumInReceiverQueue, totalMsgsReceived, totaBatchReceivedFailed, totalAcksSent, totalAcksFailed, rateMsgsReceived);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void stopThread() {
        log.info("stop pulsar consumer ...");
        while (true){
            this.stopConsumer = true;
            if (consumer != null) {
                try {
                    log.warn("stop pulsar consumer . consumerName={} is connected={}", consumer.getConsumerName(), consumer.isConnected());
                    consumer.stop();
                } catch (PulsarClientException e) {
                    e.printStackTrace();
                }
            }
            if(consumer==null||consumer.isConnected()==false){
                log.info("stop pulsar consumer finally");
                break;
            }
        }
    }


    private Object[] buildArgs(Message<String> message, Messages<String> messages, XPulsarConsumer<String> consumer) {
        int parameterCount = method.getParameterCount();
        Object[] args = new Object[parameterCount];
        Class<?>[] parameterTypes = method.getParameterTypes();
        for (int i1 = 0; i1 < parameterTypes.length; i1++) {
            Class<?> parameterType = parameterTypes[i1];
            if (parameterType.getSimpleName().equalsIgnoreCase(Message.class.getSimpleName())) {
                args[i1] = message;
            } else if (parameterType.getSimpleName().equalsIgnoreCase(Messages.class.getSimpleName())) {
                args[i1] = messages;
            } else if (parameterType.getClass() == Consumer.class.getClass()) {
                args[i1] = consumer.getConsumer();
            }
        }
        return args;
    }
}
